#include "mprpcapplication.h"
#include "rpcprovider.h"
#include "rpcheader.pb.h"
#include "zookeeperutil.h"

// 发布rpc服务方法接口
void RpcProvider::NotifyService(google::protobuf::Service* service) {
    ServiceInfo service_info;

    // 获取服务对象的信息
    const google::protobuf::ServiceDescriptor* pserviceDesc =
        service->GetDescriptor();
    // 获取服务的名字
    std::string service_name = pserviceDesc->name();
    // 获取服务对象的方法数量
    int methodCnt = pserviceDesc->method_count();

    std::cout << "service_name:" << service_name << std::endl;

    for (int i = 0; i < methodCnt; ++i) {
        // 获取服务对象指定下标的服务方法描述
        const google::protobuf::MethodDescriptor* pmethodDesc = pserviceDesc->method(i);
        std::string method_name = pmethodDesc->name();
        service_info.m_methodMap.insert(std::make_pair(method_name, pmethodDesc));

        std::cout << "method_name:" << method_name << std::endl;
    }
    service_info.m_service = service;
    m_serviceMap.insert(std::make_pair(service_name, service_info));
}

// 启动rpc节点
void RpcProvider::Run() {
    std::string ip = MprpcApplication::GetInstance().GetConfing().Load("rpcserverip");
    uint16_t port = atoi(MprpcApplication::GetInstance().GetConfing().Load("rpcserverport").c_str());
    muduo::net::InetAddress address(ip, port);

    // 创建TcpServer对象
    muduo::net::TcpServer server(&m_eventloop, address, "RpcProvider");
    // 设置回调函数
    server.setConnectionCallback(
        std::bind(&RpcProvider::OnConnection, this, std::placeholders::_1));
    server.setMessageCallback(
        std::bind(&RpcProvider::OnMessage, this, std::placeholders::_1,
                    std::placeholders::_2, std::placeholders::_3));

    // 设置工作线程数量
    server.setThreadNum(4);
    
    // 改用zookeeper
    ZkClient zkCli;
    zkCli.Start();
    // service_name为永久性节点， method_name为临时性节点
    for (auto& sp : m_serviceMap) {
        std::string service_path = "/" + sp.first;
        zkCli.Create(service_path.c_str(), nullptr, 0);
        for (auto& mp : sp.second.m_methodMap) {
            std::string method_path = service_path + "/" + mp.first;
            char method_path_data[128] = {0};
            sprintf(method_path_data, "%s:%d", ip.c_str(), port);
            zkCli.Create(method_path.c_str(), method_path_data, strlen(method_path_data), ZOO_EPHEMERAL);
        }
    }

    std::cout << "RpcProvider start service at ip:" << ip << " port:" << port << std::endl;

    // 启动网络服务
    server.start();
    m_eventloop.loop();
}

void RpcProvider::OnConnection(const muduo::net::TcpConnectionPtr& conn) {
    if (!conn->connected()) {
        // 断开连接
        conn->shutdown();
    }
}

// 已连接的用户读事件回调，来了rpc服务请求就响应
/*
 * provider 和 consumer 需要协商好之间传递的数据格式
 * server_name method_name args 利用protobuf进行数据头的序列化和反序列化
 * header_size(4个字节) + header_str(服务对象和方法) + args_str(调用方法的参数) 
 */
void RpcProvider::OnMessage(const muduo::net::TcpConnectionPtr& conn, 
                            muduo::net::Buffer* buf, 
                            muduo::Timestamp time) {
    std::string recv_buf = buf->retrieveAllAsString();
    
    // 从字符流中读取前四个字节
    uint32_t header_size = 0;
    recv_buf.copy((char*)&header_size, 4, 0);

    // 根据head_size大小读取数据头的原始字节流
    std::string rpc_header_str = recv_buf.substr(4, header_size);
    mprpc::RpcHeader rpcHeader;
    std::string service_name;
    std::string method_name;
    uint32_t args_size;
    // 解析数据头
    if(rpcHeader.ParseFromString(rpc_header_str)) {
        // 反序列化成功
        service_name = rpcHeader.service_name();
        method_name = rpcHeader.method_name();
        args_size = rpcHeader.args_size();
    }
    else {
        // 反序列化失败
        std::cerr << "rpc_header_std:" << rpc_header_str << " parse error!" << std::endl;
        return;
    }

    // 解析调用rpc方法参数
    std::string args_str = recv_buf.substr(4 + header_size, args_size);

    // 调试信息
    std::cout << "========================================" << std::endl;
    std::cout << "header_size:" << header_size << std::endl;
    std::cout << "rpc_header_str:" << rpc_header_str << std::endl;
    std::cout << "service_name:" << service_name << std::endl;
    std::cout << "method_name:" << method_name << std::endl;
    std::cout << "args_str:" << args_str << std::endl;
    std::cout << "========================================" << std::endl;

    // 获取对方rpc调用想要的service对象和method对象 通过name 找到 本地存储的对象
    auto it = m_serviceMap.find(service_name);
    if (it == m_serviceMap.end()) {
        std::cout << service_name << " is not exist!" << std::endl;
        return;
    }
    
    auto mit = it->second.m_methodMap.find(method_name);
    if(mit == it->second.m_methodMap.end()) {
        std::cout << service_name << ":" << method_name << " is not exist!" << std::endl;
        return;
    }

    google::protobuf::Service* service = it->second.m_service;
    const google::protobuf::MethodDescriptor* method = mit->second;

    // 生成rpc方法调用的请求request和响应response参数
    google::protobuf::Message* request = service->GetRequestPrototype(method).New();
    if(!request->ParseFromString(args_str)) {
        std::cout << "request parse error, content:" << args_str << std::endl;
    }
    google::protobuf::Message* response = service->GetResponsePrototype(method).New();

    // 给method方法调用绑定一个Closure参数
    google::protobuf::Closure* done = google::protobuf::NewCallback<RpcProvider,
                                                                    const muduo::net::TcpConnectionPtr&,
                                                                    google::protobuf::Message*>
                                                                    (this, &RpcProvider::SendRpcResponse, conn, response);

    service->CallMethod(method, nullptr, request, response, done);
}

void RpcProvider::SendRpcResponse(const muduo::net::TcpConnectionPtr& conn, google::protobuf::Message* response) {
    std::string response_str;
    if(response->SerializeToString(&response_str)) {
        // 序列化成功
        conn->send(response_str);
    }
    else {
        std::cout << "serialize response_str error!" << std::endl;
    }
    conn->shutdown();       // 短链接，一次后服务方主动断开连接
}